package com.huan.gateway.flow;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class MyProcessor extends SubmissionPublisher<String> implements Flow.Processor<Integer,String> {

    /**
     * 处理者和发布者的订阅关系
     */
    private Flow.Subscription subscription;
    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        //1. 保存和发布的订阅关系
        this.subscription=subscription;
        //2. 告诉发布者，我要订阅一个消息，发布者就会把消息推送到下面的 onNext 方法里面
        this.subscription.request(1);
    }

    /**
     * 处理发布者的消息
     * @param item the item
     */
    @Override
    public void onNext(Integer item) {
        //1. 处理发布者发来的数据
        System.out.println("处理发布者发来的数据：" + item);
        //2. 数据处理完成了，那么发布个最终的订阅者
        this.submit("S"+item);
        //3.处理完成数据之后，我们可以告诉发布者我还要订阅一个消息，程序就以此运行下去
        this.subscription.request(1);
    }
    /**
     *  消费消息出现异常会进到这步
     * @param throwable the exception
     */
    @Override
    public void onError(Throwable throwable) {
        System.out.println(throwable.getMessage());
        //出现异常，我们就可以告诉发布者。不接受数据了
        this.subscription.cancel();
    }

    /**
     * 在全部数据消费完成，这个方法才会调用。如果中途消费时候调用 cancel 方法，这个方法就不会触发了
     */
    @Override
    public void onComplete() {
        System.out.println("MyProcessor 处理器处理全部数据（处理器关闭了）");
    }
}
